#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../common/Log.hpp"
#include "../common/ThreadPool.hpp"
#include "../common/message.pb.h"
#include "../common/protocol.pb.h"
#include "Consumer.hpp"
#include "ConsumerManager.hpp"
#include "VirtualHost.hpp"
#include "Router.hpp"
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <mutex>
#include <functional>

namespace micromq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    using OpenChannelRequestPtr = std::shared_ptr<OpenChannelRequest>;
    using CloseChannelRequestPtr = std::shared_ptr<CloseChannelRequest>;
    using DeclareExchangeRequestPtr = std::shared_ptr<DeclareExchangeRequest>;
    using RemoveExchangeRequestPtr = std::shared_ptr<RemoveExchangeRequest>;
    using DeclareMsgQueueRequestPtr = std::shared_ptr<DeclareMsgQueueRequest>;
    using RemoveMsgQueueRequestPtr = std::shared_ptr<RemoveMsgQueueRequest>;
    using BindRequestPtr = std::shared_ptr<BindRequest>;
    using UnbindRequestPtr = std::shared_ptr<UnbindRequest>;
    using PublishMessageRequestPtr = std::shared_ptr<PublishMessageRequest>;
    using AckMessageRequestPtr = std::shared_ptr<AckMessageRequest>;
    using FollowMsgQueueRequestPtr = std::shared_ptr<FollowMsgQueueRequest>;
    using UnfollowMsgQueueRequestPtr = std::shared_ptr<UnfollowMsgQueueRequest>;

    using ConsumerResponsePtr = std::shared_ptr<ConsumerResponse>;
    using CommonResponsePtr = std::shared_ptr<CommonResponse>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
    public:
        Channel(const std::string& channelId, const VirtualHost::ptr& host, const ConsumerManager::ptr& consumerManager, 
            const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
            : _channelId(channelId), _host(host), _consumerManager(consumerManager)
            , _conn(conn), _codec(codec)
        {}

        void declareExchange(const DeclareExchangeRequestPtr& req)
        {
            bool result = _host->declareExchange(req->exchange_name(), req->echange_type(), 
                req->durable(), req->auto_delete(), req->args());
            sendCommonResponse(req->req_id(), req->channel_id(), result);
        }

        void removeExchange(const RemoveExchangeRequestPtr& req)
        {
            _host->removeExchange(req->exchange_name());
            sendCommonResponse(req->req_id(), req->channel_id(), true);
        }

        void declareMsgQueue(const DeclareMsgQueueRequestPtr& req)
        {
            bool result = _host->declareMsgQueue(req->msg_queue_name(), req->durable(),
                req->exclusive(), req->auto_delete(), req->args());
            if(result == false) sendCommonResponse(req->req_id(), req->channel_id(), result);
            _consumerManager->createConsumerCenter(req->msg_queue_name());
            sendCommonResponse(req->req_id(), req->channel_id(), result);
        }

        void removeMsgQueue(const RemoveMsgQueueRequestPtr& req)
        {
            _host->removeMsgQueue(req->msg_queue_name());
            _consumerManager->deleteConsumerCenter(req->msg_queue_name());
            sendCommonResponse(req->req_id(), req->channel_id(), true);
        }

        void bind(const BindRequestPtr& req)
        {
            if(!Router::isBindingKeyValid(req->binding_key())) 
            {
                sendCommonResponse(req->req_id(), req->channel_id(), false);
            } 
            else 
            {
                bool result = _host->bind(req->exchange_name(), req->msg_queue_name(), req->binding_key());
                sendCommonResponse(req->req_id(), req->channel_id(), result);
            }
        }

        void unbind(const UnbindRequestPtr& req)
        {
            _host->unbind(req->exchange_name(), req->msg_queue_name());
            sendCommonResponse(req->req_id(), req->channel_id(), true);
        }

        void publishMessage(const PublishMessageRequestPtr& req)
        {
            // 获取交换机
            Exchange::ptr exchange = _host->getExchangeByName(req->exchange_name());
            if(exchange.get() == nullptr)
            {
                sendCommonResponse(req->req_id(), req->channel_id(), false);
            }
            // 进行路由匹配
            BindingInfo bindingInfo = _host->getBindingInfoByEcxchange(req->exchange_name());
            BasicProperties* properties = nullptr;
            std::string routingKey;
            if(req->has_properties())
            {
                properties = req->mutable_properties();
                routingKey = properties->routing_key();
            }
            if(!Router::isRoutingKeyValid(routingKey))
            {
                sendCommonResponse(req->req_id(), req->channel_id(), false);
            }
            for(auto& it : bindingInfo) 
            {
                if(Router::route(exchange->_type, it.second->_bindingKey, routingKey))
                {
                    // 将消息添加进队列
                    _host->produceMessage(it.first, properties, req->content());
                    // 将消息推送给消费者（相当于条件变量就绪，notify）
                    ThreadPool::getInstance()->submit(std::bind(&Channel::pushMessageToConsumer, this, it.first));
                }
            }
            sendCommonResponse(req->req_id(), req->channel_id(), true);
        }

        void ackMessage(const AckMessageRequestPtr& req)
        {
            _host->ackMessage(req->msg_queue_name(), req->message_id());
            sendCommonResponse(req->req_id(), req->channel_id(), true);
        }

        void followMsgQueue(const FollowMsgQueueRequestPtr& req)
        {
            if(!_host->isMsgQueueExists(req->msg_queue_name()))
            {
                logInfo("message queue: %s is not exists, follow faild...", req->msg_queue_name().c_str());
                sendCommonResponse(req->req_id(), req->channel_id(), false);
                return;
            }
            if(_host->getMsgQueueByName(req->msg_queue_name())->_exclusive == true 
                && _consumerManager->getConsumerFromMsgQueue(req->msg_queue_name()).get() != nullptr)
            {
                sendCommonResponse(req->req_id(), req->channel_id(), false);
                return;
            }
            // 记录发起连接的客户端，所产生的消费者对象，信道关闭时将其移除
            _consumer = _consumerManager->addConsumerToMsgQueue(req->consumer_tag(), req->msg_queue_name(), req->auto_ack(), 
                std::bind(&Channel::sendConsumerResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            if(_consumer.get() == nullptr) {
                logInfo("consumer: %s followe message queue faild...", req->consumer_tag().c_str());
            } else {
                logInfo("consumer: %s followed message queue: %s", req->consumer_tag().c_str(), req->msg_queue_name().c_str());
            }
            sendCommonResponse(req->req_id(), req->channel_id(), _consumer.get() != nullptr);
        }

        void unfollowMsgQueue(const UnfollowMsgQueueRequestPtr& req)
        {
            _consumerManager->removeConsumerFromMsgQueue(req->consumer_tag(), req->msg_queue_name());
            sendCommonResponse(req->req_id(), req->channel_id(), true);
        }

        ~Channel()
        {
            if(_consumer.get() != nullptr)
            {
                _consumerManager->removeConsumerFromMsgQueue(_consumer->_tag, _consumer->_msgQueueName);
            }
        }
    private:
        void sendCommonResponse(const std::string& reqId, const std::string& channelId, bool isOk)
        {
            CommonResponse resp;
            resp.set_resp_id(reqId);
            resp.set_channel_id(channelId);
            resp.set_is_ok(isOk);
            _codec->send(_conn, resp);
        }

        void pushMessageToConsumer(const std::string& msgQueueName)
        {
            // 从队列中取出一条消息
            MessagePtr message = _host->consumeMessage(msgQueueName);
            if(message.get() == nullptr)
            {
                logInfo("message from %s is nothing...", msgQueueName.c_str());
                return;
            }
            // 从队列中取出一个消费者
            Consumer::ptr consumer = _consumerManager->getConsumerFromMsgQueue(msgQueueName);
            if(consumer.get() == nullptr)
            {
                logInfo("message queue: %s has no consumer...", msgQueueName.c_str());
                return;
            }
            // 调用消费者的消息处理回调，把消息内容及其参数发给消费者
            consumer->_callback(consumer->_tag, message->mutable_payload()->mutable_properties(), message->payload().content());
            if(consumer->_autoAck == true)
            {
                _host->ackMessage(msgQueueName, message->payload().properties().id());
            }
        }

        void sendConsumerResponse(const std::string& tag, const BasicProperties* bp, const std::string& content)
        {
            ConsumerResponse resp;
            resp.set_channel_id(_channelId);
            resp.set_consumer_tag(tag);
            resp.set_content(content);
            if(bp != nullptr) 
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_mode(bp->mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }
    private:
        std::string _channelId;
        Consumer::ptr _consumer;
        ConsumerManager::ptr _consumerManager;
        VirtualHost::ptr _host;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
    };
}